package com.aotain.nyx.apptype;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;

public class AppTypeStatMap implements MapFunction<Tuple2<String,String>,Tuple2<String,AppTypeTuple>> {

	/** 
	* @Fields serialVersionUID : TODO(用一句话描述这个变量表示什么) 
	*/ 
	private static final long serialVersionUID = -2716766572348662924L;

	@Override
	public Tuple2<String, AppTypeTuple> map(Tuple2<String,String> value)
			throws Exception {
		// TODO Auto-generated method stub
		Tuple2<String,AppTypeTuple> ret = new Tuple2<String,AppTypeTuple>();
		String[] items = value.f1.split("\\|",-1);
		if(items.length <= 10)
			return ret;
		String sip = items[2];
		String appType = items[20];
		String appId = items[21];
		long upStreamOctets = 0; //UPSTREAMOCTETS
		long upStreamPacket = 0; //UPSTREAMPACKET			
		long dnStreamOctets = 0; //DNSTREAMOCTETS
		long dnStreamPacket = 0; //DNSTREAMPACKET
		
		upStreamOctets = Long.parseLong(items[10]);
		upStreamPacket = Long.parseLong(items[11]);
		dnStreamOctets = Long.parseLong(items[15].trim().isEmpty()?"0":items[15].trim());
		dnStreamPacket = Long.parseLong(items[16].trim().isEmpty()?"0":items[16].trim());
		
		String key = String.format("%s_%s_%s", sip,appType,appId);
		AppTypeTuple tuple = new AppTypeTuple();
		tuple.setSourceIP(sip);
		tuple.setAppType(appType);
		tuple.setAppId(appId);
		tuple.setUpStreamOctets(upStreamOctets);
		tuple.setUpStreamPacket(upStreamPacket);
		tuple.setDnStreamOctets(dnStreamOctets);
		tuple.setDnStreamPacket(dnStreamPacket);
		
		ret.setFields(key, tuple);
		
		return ret;
	}

}
